package org.huatek.anders.config;

import org.huatek.anders.constant.RabbitConstant;
import org.huatek.anders.websocket.WebSocketServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;

import java.io.IOException;

/**
 * @author anders
 * @since 2018/11/18 21:06
 */
@Slf4j
@Configuration
public class RabbitReceiveConfig implements RabbitListenerConfigurer {
    private final ConnectionFactory connectionFactory;
    private final WebSocketServerEndpoint webSocketServerEndpoint;

    @Autowired
    public RabbitReceiveConfig(ConnectionFactory connectionFactory,
                               WebSocketServerEndpoint webSocketServerEndpoint){
        this.connectionFactory = connectionFactory;
        this.webSocketServerEndpoint = webSocketServerEndpoint;
    }

    @Bean
    public Queue tipsQueue() {
        return new Queue(RabbitConstant.QUEUE_TIPS, true);
    }
    /**
     * 设置JSON格式接收数据
     */
    @Bean
    public DefaultMessageHandlerMethodFactory handlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }


    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setQueueNames(RabbitConstant.QUEUE_TIPS);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            byte[] body = message.getBody();

            log.info("receive msg : " + new String(body));
            try {
                webSocketServerEndpoint.sendMessageToAll(new String(body));
                //确认消息成功消费
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                log.error("消息推送前台出错：" + e.getMessage() + "/r/n重新发送");
                //重新发送
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        });
        return container;
    }


    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
        rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(handlerMethodFactory());
    }

}
